package cn.pm.datasource.sharding;

import cn.pm.datasource.sharding.bean.ShardingRedisMessage;
import cn.pm.datasource.sharding.consts.RedisListenerTopicConst;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.HashSet;
import java.util.Set;

/**
 * Create by spf on 2020/07/28
 *
 * 用于监听更新实际表的redis广播监听器
 * @see AppConfig#redisMessageListener
 */
@Slf4j
public class ShardingRedisListener implements MessageListener {

    private final RedisSerializer<String> keySerializer = StringRedisSerializer.UTF_8;
    private final RedisSerializer valueSerializer = new JdkSerializationRedisSerializer();

    // 如果是本进程发出的广播消息 需要在这里标记一下  主要用于区分消息是不是本线程发出的
    public static final Set<String> flagToken = new HashSet<>();

    /**
     * 处理Redis订阅消息
     */
    @SneakyThrows
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String channel = keySerializer.deserialize(message.getChannel());
        if (!RedisListenerTopicConst.AUTO_CREATE_TABLE_UPDATE_TOPIC.equals(channel)) {
            log.warn("{} 监听到了未知的Redis Topic : {}", this.getClass(), channel);
            return;
        }
        Object body = valueSerializer.deserialize(message.getBody());
        if (!(body instanceof ShardingRedisMessage))
            return;
        ShardingRedisMessage bean = (ShardingRedisMessage) body;
        // 是本进程发出的 message  跳过即可
        if (flagToken.contains(bean.getToken())) {
            flagToken.remove(bean.getToken());
            return;
        }

        ShardingTableNameUtils.updateTable(bean.getLogicTableName(), bean.getRealTableName(), bean.getDataSourceCatalog());
    }
}
